package cn.pengpeng;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;


import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Set;

public class MySQLReducer extends Reducer<MySQLMapperOutputKey, MySQLMapperOutputValue, NullWritable, MySQLReducerOutputValue> {

    private NullWritable outputKey = NullWritable.get();
    private MySQLReducerOutputValue outputValue;


    @Override
    protected void reduce(MySQLMapperOutputKey key, Iterable<MySQLMapperOutputValue> values, Context context)
            throws IOException, InterruptedException {
        outputValue = new MySQLReducerOutputValue();
        outputValue.setUrl(key.getUrl());
        outputValue.setDate(key.getDate());

        Set<String> set = new HashSet<String>();
        for(MySQLMapperOutputValue value : values) {
            set.add(value.getUid());
        }
        outputValue.setUv(set.size());
        context.write(outputKey, outputValue);
    }
}